package rdd.operate;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;

import java.util.Arrays;
import java.util.List;

public class Spark50_Operate_Shuffle {
    public static void main(String[] args) throws InterruptedException {
        final SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("spark");
        final JavaSparkContext jsc = new JavaSparkContext(conf);
        final List<Integer> nums = Arrays.asList(1,2,3,4);
        final JavaRDD<Integer> rdd = jsc.parallelize(nums , 2);
        rdd.groupBy(
                num -> num % 2 == 0
        ).collect().forEach(System.out::println);
        System.out.println("计算完毕");
        Thread.sleep(1000000L);                 //进入http://localhost:4040查看，点击collect at Spark50_Operate_Shuffle，Shuffle Read代表落盘，Shuffle Write代表写出到内存
        jsc.close();                                  //写和读都有两个Task，是因为在分区的时候，分了两个区，每个区都需要创建一个Task。就算分区为空，也需要创建Task，所以groupBy可能会产生资源浪费

        /*
        groupBy具有改变分区数量的能力
        final JavaRDD<Integer> rdd = jsc.parallelize(nums , 3);         //这里设置了三个分区
        rdd.groupBy(
                num -> num % 2 == 0
        ,2).collect().forEach(System.out::println);                    //groupBy第二个参数可以修改分区
        System.out.println("计算完毕");
        Thread.sleep(1000000L);                 //进入http://localhost:4040查看，点击collect at Spark50_Operate_Shuffle，Shuffle Read代表落盘，Shuffle Write代表写出到内存
        jsc.close();                            //可以发现在Write时候，有三个Task代表有三个分区。在Read的时候有两个Task，代表两个分区，说明Shuffle改变了分区数量
        */
    }

}
